package com.lry.service.impl;

import com.lry.core.Log;
import com.lry.core.RaftNode;
import com.lry.entity.AppendEntriesRequest;
import com.lry.entity.AppendEntriesResponse;
import com.lry.entity.VoteRequest;
import com.lry.entity.VoteResponse;
import com.lry.service.RaftConsensusService;
import lombok.extern.slf4j.Slf4j;

import java.util.List;

@Slf4j
public class RaftConsensusServiceImpl implements RaftConsensusService {

    private RaftNode raftNode;

    public RaftConsensusServiceImpl(RaftNode raftNode){
        this.raftNode = raftNode;
    }

    @Override
    public VoteResponse requestVote(VoteRequest voteRequest) {
        //如果我的term大于请求的term，则拒绝
        if(raftNode.getCurrentTerm()>voteRequest.getTerm()){
            return new VoteResponse(raftNode.getCurrentTerm(),false);
        }

        if(null==raftNode.getVotedFor()){
            synchronized (RaftConsensusServiceImpl.class){
                if(null==raftNode.getVotedFor()){
                    raftNode.setVotedFor(voteRequest.getCandidateId());
                    return new VoteResponse(raftNode.getCurrentTerm(),true);
                }
            }
        }

        return new VoteResponse(raftNode.getCurrentTerm(),false);
    }

    @Override
    public AppendEntriesResponse get(AppendEntriesRequest appendEntriesRequest) {
        log.info("server={},收到serverId={}的 get",raftNode.getServerId(),appendEntriesRequest.getLeaderId());

        Log[] logs = appendEntriesRequest.getEntries();
        if(null!=logs && logs.length>0){
            for (Log log1 : logs) {
                log.info("server={},get op={}",raftNode.getServerId(),new String(log1.getCommand()));
                raftNode.getStateMachine().get(log1.getCommand());
            }
        }
        return new AppendEntriesResponse(raftNode.getCurrentTerm(),true);
    }
    @Override
    public AppendEntriesResponse appendEntries(AppendEntriesRequest appendEntriesRequest) {
        log.info("server={},收到leaderId={}的 appendEntries",raftNode.getServerId(),appendEntriesRequest.getLeaderId());

        if (null==raftNode.getLeaderId()) {
            raftNode.setLeaderId(appendEntriesRequest.getLeaderId());
        }

        //收到leader的追加日志或者心跳，重置自己的选举计时器
        //raftNode.resetElectionTimer(); //先注释掉，不给他篡权，为了调试日志方便

        //如果我的term大于请求的term，则拒绝
        if(raftNode.getCurrentTerm()>appendEntriesRequest.getTerm()){
            return new AppendEntriesResponse(raftNode.getCurrentTerm(),false);
        }
        //心跳
        if(appendEntriesRequest.getEntries()==null){
            return new AppendEntriesResponse(raftNode.getCurrentTerm(),true);
        }

        long preLogIndex = appendEntriesRequest.getPreLogIndex();
        long preLogTerm = appendEntriesRequest.getPreLogTerm();
        Log[] logs = appendEntriesRequest.getEntries();
        List<Log> existLogs = raftNode.getLogs();

        //在接收者的日志中查找看看是否可以找到和pre一样的日志，没有的话，说明当前节点数据落后了，需要返回false，等待leader追加过来
        if(null!=logs&& logs.length>0&&!existLogs.isEmpty()&&preLogIndex!=0&&preLogTerm!=0){
            boolean findPre = false;
            for(int i=existLogs.size()-1;i>=0;i--){
                Log log = existLogs.get(i);
                if(log.getIndex()==preLogIndex&&log.getTerm()==preLogTerm){
                    findPre = true;
                    break;
                }
            }
            if(!findPre){
                return new AppendEntriesResponse(raftNode.getCurrentTerm(),false);
            }
        }

        //如果一个已经存在的条目和新条目发生了冲突（索引相同，任期不同），那么就删除这个已经存在条目及它之后的所有条目
        if(null!=logs&& logs.length>0&&!existLogs.isEmpty()){
            //已经存在的条目和新条目 最小冲突的索引
            int minConflictIndex = Integer.MAX_VALUE;
            for (Log log1 : logs) {
                for(int i=existLogs.size()-1;i>=0;i--){
                    Log log = existLogs.get(i);
                    if(log.getIndex()==log1.getIndex()&&log.getTerm()!=log1.getTerm()){
                        minConflictIndex = Math.min(minConflictIndex,i);
                    }
                }
            }
            //移除冲突之后的条目
            for(int i = minConflictIndex;i<existLogs.size();i++){
                existLogs.remove(existLogs.get(i));
            }
        }

        if(null!=logs && logs.length>0){
            for (Log log1 : logs) {
                log.info("server={},追加日志={}",raftNode.getServerId(),new String(log1.getCommand()));
                raftNode.getLogs().add(log1);
                raftNode.getStateMachine().apply(log1.getCommand());
            }
            if(appendEntriesRequest.getLeaderCommit()>raftNode.getCommitIndex()){
                raftNode.setCommitIndex(Math.min(appendEntriesRequest.getLeaderCommit(),raftNode.getLastLogIndex()));
            }
        }

        return new AppendEntriesResponse(raftNode.getCurrentTerm(),true);
    }
}
